package cn.itcast.up.model.matchtag

import java.util.Properties

import cn.itcast.up.model.bean.{HBaseMeta, TagRule}
import org.apache.spark.sql.{DataFrame, Dataset, Row, SparkSession}

/**
  * 性别模型标签. cn.itcast.up.model.matchtag.GenderModel
  */
object GenderModel {
  def main(args: Array[String]): Unit = {
    //1. 构建一个SparkSession的对象
    val spark: SparkSession = SparkSession.builder()
      .master("local[*]")
      .appName("GenderModel")
      .getOrCreate()
    import spark.implicits._
    import org.apache.spark.sql.functions._


    //2. 加载MySQL数据源,整个数据库中的表数据都可以获取到了.
    //定义连接MySQL的字段
    //URL后面追加的有MySQL的用户名密码.
    val url = "jdbc:mysql://bd001:3306/tags_new?useUnicode=true&characterEncoding=UTF-8&serverTimezone=UTC&user=root&password=123456"
    val tableName = "tbl_basic_tag"
    val props = new Properties


    val mysqlSource: DataFrame = spark.read
      //      .format("jdbc")
      //      .option("url", "jdbc....")
      .jdbc(url, tableName, props)
//    mysqlSource.show()

    /**
      * +---+----+--------+--------------------+--------+-----+---+-----+-----+-----+------+
      * | id|name|industry|                rule|business|level|pid|ctime|utime|state|remark|
      * +---+----+--------+--------------------+--------+-----+---+-----+-----+-----+------+
      * |469|  电商|    null|                null|    null|    1| -1| null| null| null|  null|
      * |470|京东商城|    null|                null|    null|    2|469| null| null| null|  null|
      * |471|用户属性|    null|                null|    null|    3|470| null| null| null|  null|
      * |472|  性别|    null|inType=HBase##zkH...|    null|    4|471| null| null| null|  null|
      * |473|  保密|    null|                   0|    null|    5|472| null| null| null|  null|
      * |474|   男|    null|                   1|    null|    5|472| null| null| null|  null|
      * |475|   女|    null|                   2|    null|    5|472| null| null| null|  null|
      * +---+----+--------+--------------------+--------+-----+---+-----+-----+-----+------+
      */

    //3. 获取tbl_basic_tag中的数据:
    //   1. 4级标签的数据:id,rule:数据源信息
    val fourDS: Dataset[Row] = mysqlSource.select('id, 'rule)
      .where("id = 472")

//      .show()
//    +---+--------------------+
    //| id|                rule|
    //+---+--------------------+
    //|472|inType=HBase##zkH...|
    //+---+--------------------+
    //将4级的数据转换为一个对象HBaseMeta对象.
    val fourMap: Map[String, String] = fourDS.map(srcSource => {
      //将row中的数据获取出来
      //inType=HBase##zkHosts=192.168.10.20##zkPort=2181##hbaseTable=tbl_users##family=detail##selectFields=id,gender
      val ruleStr: String = srcSource.getAs[String]("rule")
      //## 切割
      ruleStr.split("##")
        .map(kv => {
          //= 切割
          val tmpArr: Array[String] = kv.split("=")
          (tmpArr(0), tmpArr(1))
        })
      //      Array[(String, String)]
    })
      //将数据采集为集合,并且取第一个元素,这个就是我们的4级的规则数据
      .collect()(0).toMap
//    println(fourMap)
    //将4级数据转换为HBaseMeta对象
      val meta = HBaseMeta(fourMap)
    //   2. 5级标签的数据:id,rule: 匹配的时候使用.
    //根据pid加载5级规则数据
    val fiveDs: Dataset[Row] = mysqlSource.select('id, 'rule)
      .where("pid == 472")
    val fiveRules: List[TagRule] = fiveDs.map(row => {
      //获取id
      val id: String = row.getAs[Long]("id").toString
      //获取rule
      val rule: String = row.getAs[String]("rule")
      TagRule(id, rule)
    }).collect().toList
//    println(fiveRules)
//    List(TagRule(473,0), TagRule(474,1), TagRule(475,2))

    //val 数据源

    //先判断当前的数据源在哪?
    if (meta.inType.equalsIgnoreCase("hbase")) {
      //从HBase中加载数据源
    } else if (meta.inType.equalsIgnoreCase("hive")){
      //从Hive中获取
    }


    //4. 获取HBase中的数据
    //   1. 使用自定义数据源加载HBase中的数据.
    //Hbase数据只能通过API来进行加载
        //1. 编写一个Hbaseutils工具类.手动加载数据,效率低
        //2. 使用自定义数据源的方式进行加载.
    val hbaseSource: DataFrame = spark.read
      .format("cn.itcast.up.model.HBaseSource")
      .option(HBaseMeta.ZKHOSTS, meta.zkHosts)
      .option(HBaseMeta.ZKPORT, meta.zkPort)
      .option(HBaseMeta.HBASETABLE, meta.hbaseTable)
      .option(HBaseMeta.FAMILY, meta.family)
      .option(HBaseMeta.SELECTFIELDS, meta.selectFields)
      .load()
//    hbaseSource.show()
//    +---+---+------+
    //|row| id|gender|
    //+---+---+------+
    //|  1|  1|     2|
    //| 10| 10|     2|
    //|100|100|     2|

    //5. 使用5级规则数据和HBase中的源数据进行匹配,找到用户所属的标签.
    //定义一个自定义函数,通过gender获取对应的5级id

    val getTagId = udf((gender: String) =>{
      //判断当前gender属于哪个5级,并返回5级id
      var tagID:String = ""
      for (tagRule <- fiveRules) {
        //判断是否相等
        if (tagRule.rule.equals(gender)){
          tagID = tagRule.id
        }
      }
      //将匹配的ID返回
      tagID
    })


    val resultFrame: DataFrame = hbaseSource.select('id.as("userid"), getTagId('gender).as("tagId"))

//    resultFrame.show()
//    +------+-----+
//    |userid|tagId|
//    +------+-----+
//    |     1|  475|
//      |    10|  475|
//      |   100|  475|
//      |   101|  474|
//      |   102|  475|



    //6. 将结果存入HBase.
  }
}
